package com.bigdata.flink.kafkaproducer;

import com.bigdata.flink.conf.KafkaProperties;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Random;

public class KafkaProducerExample {

    private static final Logger logger = LogManager.getLogger(KafkaProducerExample.class);

    private static String topicname = KafkaProperties.getTopic();

    public static void main(String[] args) throws InterruptedException {

        final Producer<String, String> producer = KafkaProperties.createProducer();

        while (true){
            StringBuilder builder = new StringBuilder();

            builder.append("imooc").append("\t")
                    .append("CN").append("\t")
                    .append(getLevels()).append("\t")
                    .append(new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date())).append("\t")
                    .append(getIps()).append("\t")
                    .append(getDomains()).append("\t")
                    .append(getTraffic()).append("\t");
            //System.out.println(builder.toString());
            producer.send(new ProducerRecord<>(topicname, builder.toString()));
            logger.info("Sent {} to topic {}", builder.toString(), topicname);

            Thread.sleep(2000);
        }

    }

    //生产level数据
    private static String getLevels(){
        String[] levels = new String[]{"M", "E"};
        return levels[new Random().nextInt(levels.length)];
    }

    //生产ip数据
    private static String getIps(){
        String[] ips = new String[]{
                "223.104.18.110",
                "113.101.75.194",
                "27.17.127.135",
                "183.225.139.16",
                "112.1.66.34",
                "175.148.211.190",
                "183.227.58.21",
                "59.83.198.84",
                "117.28.38.28",
                "117.59.39.169"
        };
        return ips[new Random().nextInt(ips.length)];
    }

    private static String getDomains(){
        String[] domains = new String[]{
                "v1.go2yd.com",
                "v2.go2yd.com",
                "v3.go2yd.com",
                "v4.go2yd.com",
                "vmi.go2yd.com"
        };
        return domains[new Random().nextInt(domains.length)];
    }

    private static long getTraffic(){
        return new Random().nextInt(10000);
    }


}
